package com.atbeijing.D05;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;


/**
 * 每5s报告一次某个key流过了多少数据
 */
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .filter(r -> r.id.equals("sensor_1"))
                .keyBy(r -> r.id)
                .process( new MyKeyedProcessFunction1())
                .print();

        env.execute();
    }

    public static class MyKeyedProcessFunction1 extends KeyedProcessFunction<String, SensorReading,String>{
        private ListState<SensorReading> listState;
        private ValueState<Long> timeTs;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            listState=getRuntimeContext().getListState(new ListStateDescriptor<SensorReading>("list-state",SensorReading.class));
            timeTs=getRuntimeContext().getState(new ValueStateDescriptor<Long>("time-ts", Types.LONG));
        }

        @Override
        public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
            listState.add(value);
            //定时器不为空说明在计时
            if (timeTs.value()==null){
                //注册定时器:5s后
                long timeLater = ctx.timerService().currentProcessingTime() + 5000L;
                ctx.timerService().registerProcessingTimeTimer(timeLater);
                timeTs.update(timeLater);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);

            //输出结果后清空重新计时
            out.collect("共"+listState.get().spliterator().getExactSizeIfKnown()+"条数据");
            timeTs.clear();
        }
    }
}
